﻿#include "kafka_producer.h"

// 生产者回调类
class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message& message) override {
        if (message.err() == RdKafka::ERR__TIMED_OUT) {
            std::cout << "Message timed out after " << message.latency() << " ms" << std::endl;
        }
        else if (message.err()) {
            std::cout << "Delivery failed: " << RdKafka::err2str(message.err()) << std::endl;
        }
        else {
            std::cout << "Message delivered to topic " << message.topic_name() << " [" << message.partition() << "] at offset " << message.offset() << std::endl;
        }
    }
};

// 生产者事件回调类
class ExampleEventCb : public RdKafka::EventCb {
public:
    void event_cb(RdKafka::Event& event) override {
        switch (event.type()) {
        case RdKafka::Event::EVENT_ERROR:
            std::cerr << "ERROR (" << RdKafka::err2str(event.err()) << "): " << event.str() << std::endl;
            if (event.err() == RdKafka::ERR__FATAL)
                exit(1);
            break;
        case RdKafka::Event::EVENT_STATS:
            std::cerr << "\"STATS\": " << event.str() << std::endl;
            break;
        case RdKafka::Event::EVENT_LOG:
            fprintf(stderr, "LOG-%i-%s: %s\n", event.severity(), event.fac().c_str(), event.str().c_str());
            break;
        default:
            std::cerr << "EVENT " << event.type() << " (" << RdKafka::err2str(event.err()) << "): " << event.str() << std::endl;
            break;
        }
    }
};

ExampleProducer::ExampleProducer(const std::string& brokers, const std::string& topic_name) {
    // 创建配置对象
    conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("bootstrap.servers", brokers, errstr);
    conf->set("dr_cb", new ExampleDeliveryReportCb(), errstr);
    conf->set("event_cb", new ExampleEventCb(), errstr);

    // 创建生产者对象
    producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        exit(1);
    }

    // 创建主题对象
    topic = RdKafka::Topic::create(producer, topic_name, nullptr, errstr);
    if (!topic) {
        std::cerr << "Failed to create topic: " << errstr << std::endl;
        exit(1);
    }
}

ExampleProducer::~ExampleProducer() {
    delete topic;
    delete producer;
    delete conf;
}

void ExampleProducer::produce(const std::string& message) {
    RdKafka::ErrorCode err = producer->produce(topic, RdKafka::Topic::PARTITION_UA, RdKafka::Producer::RK_MSG_COPY, const_cast<char*>(message.c_str()), message.length(), NULL, NULL);
    if (err != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Produce failed: " << RdKafka::err2str(err) << std::endl;
    }
    else {
        std::cout << "Message produced" << std::endl;
    }
    try
    {
        producer->poll(0);
        // 等待所有消息发送完成
        producer->flush(1000 * 2);
    }
    catch (const std::exception&)
    {

    }   
}